{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 1,
   "id": "b2236469-9a18-4927-bf36-f6ebc500ed42",
   "metadata": {},
   "outputs": [],
   "source": [
    "import os\n",
    "import findspark\n",
    "from pyspark.sql import SparkSession\n",
    "from functools import reduce"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "id": "6b5d9566",
   "metadata": {},
   "outputs": [],
   "source": [
    "findspark.init()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "id": "a7d93cd7-2eab-4b7c-8561-57ed9f42294a",
   "metadata": {},
   "outputs": [],
   "source": [
    "spark  = SparkSession.builder.master(\"local[*]\").appName(\"windows_testing\").getOrCreate()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "id": "4993e80f-1191-4e36-836d-ca92b57b5942",
   "metadata": {},
   "outputs": [],
   "source": [
    "# folder_path = r\"D:\\bigdata\\spark_ practice\\DataAnalysisWithPythonAndPySpark-Data-trunk\\gsod_noaa\"\n",
    "# #测试环境不允许，不得已使用了reduce去拼接数据\n",
    "# list_dir = os.listdir(folder_path)\n",
    "# file_list =  [os.path.join(folder_path,file_name) for file_name in list_dir]\n",
    "# gsod= reduce(lambda x,y:x.unionByName(y,allowMissingColumns=True),[spark.read.parquet(file_path) for file_path in file_list])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "id": "c13ff7fd",
   "metadata": {},
   "outputs": [],
   "source": [
    "folder_path = r\"D:\\python work\\learning\\data\\DataAnalysisWithPythonAndPySpark-Data-trunk\\window\\gsod.parquet\"\n",
    "gsod = spark.read.parquet(folder_path)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "id": "2871d381-9aeb-440a-9a33-0f96d369de56",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------+-----+----+---+---+----+----------+------+----------+------+---------+------+---------+-----+-----------+----+----------+-----+-----+----+--------+----+--------+----+---------+-----+---+------------+----------------+----+-------+--------------------+\n",
      "|   stn| wban|year| mo| da|temp|count_temp|  dewp|count_dewp|   slp|count_slp|   stp|count_stp|visib|count_visib|wdsp|count_wdsp|mxpsd| gust| max|flag_max| min|flag_min|prcp|flag_prcp| sndp|fog|rain_drizzle|snow_ice_pellets|hail|thunder|tornado_funnel_cloud|\n",
      "+------+-----+----+---+---+----+----------+------+----------+------+---------+------+---------+-----+-----------+----+----------+-----+-----+----+--------+----+--------+----+---------+-----+---+------------+----------------+----+-------+--------------------+\n",
      "|994979|99999|2017| 12| 11|21.3|        21|9999.9|         0|1014.9|       13|9999.9|        0|999.9|          0|12.0|        13| 21.0|999.9|27.3|       *|17.1|       *| 0.0|        I|999.9|  0|           0|               0|   0|      0|                   0|\n",
      "|998012|99999|2017| 03| 02|31.4|        24|9999.9|         0|1019.4|       14|9999.9|        0|999.9|          0|15.5|        14| 24.1|999.9|33.6|       *|29.7|       *| 0.0|        I|999.9|  0|           0|               0|   0|      0|                   0|\n",
      "|719200|99999|2017| 10| 09|60.5|        11|9999.9|         0|1016.6|       11|1016.5|        8|999.9|          0|11.0|        11| 21.0|999.9|66.6|        |58.8|        | 0.0|        G|  0.4|  0|           0|               0|   0|      0|                   0|\n",
      "|998258|99999|2017| 05| 01|44.9|        13|9999.9|         0|1022.3|       13|9999.9|        0|999.9|          0| 4.6|        13|  8.0|999.9|48.7|       *|42.1|       *| 0.0|        I|999.9|  0|           0|               0|   0|      0|                   0|\n",
      "|997737|99999|2017| 03| 10| 6.8|        24|9999.9|         0|1032.9|       15|9999.9|        0|999.9|          0| 8.1|        15| 14.0|999.9|19.0|       *|-3.8|       *| 0.0|        I|999.9|  0|           0|               0|   0|      0|                   0|\n",
      "+------+-----+----+---+---+----+----------+------+----------+------+---------+------+---------+-----+-----------+----+----------+-----+-----+----+--------+----+--------+----+---------+-----+---+------------+----------------+----+-------+--------------------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "gsod.show(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "id": "99587721-2b92-4880-8e6e-ef262fb72f8b",
   "metadata": {},
   "outputs": [],
   "source": [
    "import pyspark.sql.functions as F\n",
    "import pyspark.sql.types as T"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "id": "996dd1f9-83e6-4b46-9a0d-6efc5c878986",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----+------+\n",
      "|year|  temp|\n",
      "+----+------+\n",
      "|2017|-114.7|\n",
      "|2018|-113.5|\n",
      "|2019|-114.7|\n",
      "+----+------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "coldest_temp = gsod.groupby(\"year\").agg(F.min(\"temp\").alias(\"temp\"))\n",
    "coldest_temp.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "id": "71b7cfda-404b-46e7-b34a-0cd2b0e4b938",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------+----+---+---+------+\n",
      "|   stn|year| mo| da|  temp|\n",
      "+------+----+---+---+------+\n",
      "|896060|2018| 08| 27|-113.5|\n",
      "|895770|2019| 06| 15|-114.7|\n",
      "|896250|2017| 06| 20|-114.7|\n",
      "+------+----+---+---+------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "coldest_when = (\n",
    "    coldest_temp\n",
    "    .join(gsod,on=[\"year\",\"temp\"],how=\"inner\")\n",
    "    .select(\"stn\",\"year\",\"mo\",\"da\",\"temp\")\n",
    ")\n",
    "coldest_when.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "92e2eeef-b405-4659-941b-d5d66baf33eb",
   "metadata": {},
   "source": [
    "测试一下left_semi的链接方式"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "id": "74f799d1-76a8-4992-9a15-baf385fa9d75",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------+----+---+---+------+\n",
      "|   stn|year| mo| da|  temp|\n",
      "+------+----+---+---+------+\n",
      "|896060|2018| 08| 27|-113.5|\n",
      "|895770|2019| 06| 15|-114.7|\n",
      "|896250|2017| 06| 20|-114.7|\n",
      "+------+----+---+---+------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "(\n",
    "    gsod.join(coldest_temp,on=[\"year\",\"temp\"],how=\"left_semi\")\n",
    "    .select(\"stn\",\"year\",\"mo\",\"da\",\"temp\")\n",
    "    .show()\n",
    ")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "88058d0b-f07a-49ab-b1f6-02b36862f680",
   "metadata": {},
   "source": [
    "看起来和inner没啥区别，速度更快一些，不懂是不是错觉，会按照左表的列去选择？"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "id": "70393e0b-e829-4565-95d4-07342b04c0a2",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "<pyspark.sql.window.WindowSpec object at 0x000001E3FB6EB460>\n"
     ]
    }
   ],
   "source": [
    "from pyspark.sql.window import Window\n",
    "\n",
    "each_year = Window.partitionBy(\"year\")\n",
    "print(each_year)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "id": "965d161d-f21f-4f41-91fb-dc42227c1269",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------+----+---+---+------+\n",
      "|   stn|year| mo| da|  temp|\n",
      "+------+----+---+---+------+\n",
      "|895770|2019| 06| 15|-114.7|\n",
      "|896250|2017| 06| 20|-114.7|\n",
      "|896060|2018| 08| 27|-113.5|\n",
      "+------+----+---+---+------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "(gsod\n",
    " .withColumn(\"min_temp\",F.min(\"temp\").over(each_year))\n",
    " .where(\"min_temp=temp\")\n",
    " .select(\"stn\",\"year\",\"mo\",\"da\",\"temp\")\n",
    " .show()\n",
    ")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "ca73fbff-c634-43b4-9029-9f3679d4fcc1",
   "metadata": {},
   "source": [
    "`partition_obj = Window.partitionBy(*col)`\n",
    "通过传入1列或者多列来创建窗口分区对象，将来传入over方法进行分区"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "id": "f5ffb6fd-7046-4459-885b-420f00c44a44",
   "metadata": {
    "scrolled": true
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "['stn',\n",
       " 'wban',\n",
       " 'year',\n",
       " 'mo',\n",
       " 'da',\n",
       " 'temp',\n",
       " 'count_temp',\n",
       " 'dewp',\n",
       " 'count_dewp',\n",
       " 'slp',\n",
       " 'count_slp',\n",
       " 'stp',\n",
       " 'count_stp',\n",
       " 'visib',\n",
       " 'count_visib',\n",
       " 'wdsp',\n",
       " 'count_wdsp',\n",
       " 'mxpsd',\n",
       " 'gust',\n",
       " 'max',\n",
       " 'flag_max',\n",
       " 'min',\n",
       " 'flag_min',\n",
       " 'prcp',\n",
       " 'flag_prcp',\n",
       " 'sndp',\n",
       " 'fog',\n",
       " 'rain_drizzle',\n",
       " 'snow_ice_pellets',\n",
       " 'hail',\n",
       " 'thunder',\n",
       " 'tornado_funnel_cloud']"
      ]
     },
     "execution_count": 13,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "gsod.columns"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "id": "0c2cd8a4-5610-4439-a529-7540501fa5f9",
   "metadata": {},
   "outputs": [],
   "source": [
    "# gsod_light = (\n",
    "#     gsod\n",
    "#     .withColumn(\"min_temp\",F.min(\"temp\").over(each_year))\n",
    "#     .where(\"min_temp=temp\")\n",
    "#     .select(\"stn\",\"year\",\"mo\",\"da\",\"temp\",\"count_temp\")\n",
    "# )"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "id": "8f4b707e-57f9-446f-8041-fb28e44c19dc",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------+----+---+---+----+----------+\n",
      "|   stn|year| mo| da|temp|count_temp|\n",
      "+------+----+---+---+----+----------+\n",
      "|994979|2017| 12| 11|21.3|        21|\n",
      "|998012|2017| 03| 02|31.4|        24|\n",
      "|719200|2017| 10| 09|60.5|        11|\n",
      "|917350|2018| 04| 21|82.6|         9|\n",
      "|076470|2018| 06| 07|65.0|        24|\n",
      "|996470|2018| 03| 12|55.6|        12|\n",
      "|041680|2019| 02| 19|16.1|        15|\n",
      "|949110|2019| 11| 23|54.9|        14|\n",
      "|998252|2019| 04| 18|44.7|        11|\n",
      "|998166|2019| 03| 20|34.8|        12|\n",
      "+------+----+---+---+----+----------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "folder_path = r\"D:\\python work\\learning\\data\\DataAnalysisWithPythonAndPySpark-Data-trunk\\window\\gsod_light.parquet\"\n",
    "gsod_light = spark.read.parquet(folder_path)\n",
    "gsod_light.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "30ebe34c-daa0-494d-9fac-f3b1f34b48ad",
   "metadata": {},
   "source": [
    "排序函数\n",
    "\n",
    "- rank：非连续排名\n",
    "- dense_rank：连续排名\n",
    "- percent_rank：百分位数排名\n",
    "- ntile：平铺操作\n",
    "- row_number:裸行号"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "id": "88ea31b8-a7e9-41af-bf2d-814952953e5d",
   "metadata": {},
   "outputs": [],
   "source": [
    "temp_per_month_asc = Window.partitionBy(\"mo\").orderBy(\"count_temp\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "id": "dc76199e-25d7-4288-b0e5-8673f728291d",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------+----+---+---+----+----------+--------+\n",
      "|   stn|year| mo| da|temp|count_temp|rank_tpm|\n",
      "+------+----+---+---+----+----------+--------+\n",
      "|041680|2019| 02| 19|16.1|        15|       1|\n",
      "|996470|2018| 03| 12|55.6|        12|       1|\n",
      "|998166|2019| 03| 20|34.8|        12|       1|\n",
      "|998012|2017| 03| 02|31.4|        24|       3|\n",
      "|917350|2018| 04| 21|82.6|         9|       1|\n",
      "|998252|2019| 04| 18|44.7|        11|       2|\n",
      "|076470|2018| 06| 07|65.0|        24|       1|\n",
      "|719200|2017| 10| 09|60.5|        11|       1|\n",
      "|949110|2019| 11| 23|54.9|        14|       1|\n",
      "|994979|2017| 12| 11|21.3|        21|       1|\n",
      "+------+----+---+---+----+----------+--------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "gsod_light.withColumn(\"rank_tpm\",F.rank().over(temp_per_month_asc)).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "id": "3fb038de-0f7e-45f8-a37b-66c712c1c79d",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------+----+---+---+----+----------+--------+\n",
      "|   stn|year| mo| da|temp|count_temp|rank_tpm|\n",
      "+------+----+---+---+----+----------+--------+\n",
      "|041680|2019| 02| 19|16.1|        15|       1|\n",
      "|996470|2018| 03| 12|55.6|        12|       1|\n",
      "|998166|2019| 03| 20|34.8|        12|       1|\n",
      "|998012|2017| 03| 02|31.4|        24|       2|\n",
      "|917350|2018| 04| 21|82.6|         9|       1|\n",
      "|998252|2019| 04| 18|44.7|        11|       2|\n",
      "|076470|2018| 06| 07|65.0|        24|       1|\n",
      "|719200|2017| 10| 09|60.5|        11|       1|\n",
      "|949110|2019| 11| 23|54.9|        14|       1|\n",
      "|994979|2017| 12| 11|21.3|        21|       1|\n",
      "+------+----+---+---+----+----------+--------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "gsod_light.withColumn(\"rank_tpm\",F.dense_rank().over(temp_per_month_asc)).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "id": "e6183d8d-511f-444a-9a89-73b5855e6c21",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------+----+---+---+----+----------+------------------+\n",
      "|   stn|year| mo| da|temp|count_temp|          rank_tpm|\n",
      "+------+----+---+---+----+----------+------------------+\n",
      "|994979|2017| 12| 11|21.3|        21|               0.0|\n",
      "|998012|2017| 03| 02|31.4|        24|               0.5|\n",
      "|719200|2017| 10| 09|60.5|        11|               1.0|\n",
      "|996470|2018| 03| 12|55.6|        12|               0.0|\n",
      "|076470|2018| 06| 07|65.0|        24|               0.5|\n",
      "|917350|2018| 04| 21|82.6|         9|               1.0|\n",
      "|041680|2019| 02| 19|16.1|        15|               0.0|\n",
      "|998166|2019| 03| 20|34.8|        12|0.3333333333333333|\n",
      "|998252|2019| 04| 18|44.7|        11|0.6666666666666666|\n",
      "|949110|2019| 11| 23|54.9|        14|               1.0|\n",
      "+------+----+---+---+----+----------+------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "temp_per_year = Window.partitionBy(\"year\").orderBy(\"temp\")\n",
    "gsod_light.withColumn(\"rank_tpm\",F.percent_rank().over(temp_per_year)).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "id": "f692206e-272b-49cb-9be8-07ce53e85c8d",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------+----+---+---+----+----------+--------+\n",
      "|   stn|year| mo| da|temp|count_temp|rank_tpm|\n",
      "+------+----+---+---+----+----------+--------+\n",
      "|994979|2017| 12| 11|21.3|        21|       1|\n",
      "|998012|2017| 03| 02|31.4|        24|       1|\n",
      "|719200|2017| 10| 09|60.5|        11|       2|\n",
      "|996470|2018| 03| 12|55.6|        12|       1|\n",
      "|076470|2018| 06| 07|65.0|        24|       1|\n",
      "|917350|2018| 04| 21|82.6|         9|       2|\n",
      "|041680|2019| 02| 19|16.1|        15|       1|\n",
      "|998166|2019| 03| 20|34.8|        12|       1|\n",
      "|998252|2019| 04| 18|44.7|        11|       2|\n",
      "|949110|2019| 11| 23|54.9|        14|       2|\n",
      "+------+----+---+---+----+----------+--------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "gsod_light.withColumn(\"rank_tpm\",F.ntile(2).over(temp_per_year)).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "id": "9ce370d6-4a2e-4b5b-ae7d-c5459cb4d48c",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------+----+---+---+----+----------+--------+\n",
      "|   stn|year| mo| da|temp|count_temp|rank_tpm|\n",
      "+------+----+---+---+----+----------+--------+\n",
      "|994979|2017| 12| 11|21.3|        21|       1|\n",
      "|998012|2017| 03| 02|31.4|        24|       2|\n",
      "|719200|2017| 10| 09|60.5|        11|       3|\n",
      "|996470|2018| 03| 12|55.6|        12|       1|\n",
      "|076470|2018| 06| 07|65.0|        24|       2|\n",
      "|917350|2018| 04| 21|82.6|         9|       3|\n",
      "|041680|2019| 02| 19|16.1|        15|       1|\n",
      "|998166|2019| 03| 20|34.8|        12|       2|\n",
      "|998252|2019| 04| 18|44.7|        11|       3|\n",
      "|949110|2019| 11| 23|54.9|        14|       4|\n",
      "+------+----+---+---+----+----------+--------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "gsod_light.withColumn(\"rank_tpm\",F.row_number().over(temp_per_year)).show()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "0662e038-2872-4882-a910-c005697d260a",
   "metadata": {},
   "source": [
    "分析函数\n",
    "\n",
    "- lag(col,n,default=none):给定特定记录之前的n个值\n",
    "- lead(col,n,default=none):给定特定记录之后的n个值\n",
    "- cume_dist():记录累积分布"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "id": "d3835c22-c253-4692-8469-6f9d6972246f",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------+----+---+---+----+----------+--------+---------+\n",
      "|   stn|year| mo| da|temp|count_temp|rank_tpm|rank_tpm2|\n",
      "+------+----+---+---+----+----------+--------+---------+\n",
      "|994979|2017| 12| 11|21.3|        21|    NULL|     NULL|\n",
      "|998012|2017| 03| 02|31.4|        24|    21.3|     NULL|\n",
      "|719200|2017| 10| 09|60.5|        11|    31.4|     21.3|\n",
      "|996470|2018| 03| 12|55.6|        12|    NULL|     NULL|\n",
      "|076470|2018| 06| 07|65.0|        24|    55.6|     NULL|\n",
      "|917350|2018| 04| 21|82.6|         9|    65.0|     55.6|\n",
      "|041680|2019| 02| 19|16.1|        15|    NULL|     NULL|\n",
      "|998166|2019| 03| 20|34.8|        12|    16.1|     NULL|\n",
      "|998252|2019| 04| 18|44.7|        11|    34.8|     16.1|\n",
      "|949110|2019| 11| 23|54.9|        14|    44.7|     34.8|\n",
      "+------+----+---+---+----+----------+--------+---------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "(\n",
    "    gsod_light\n",
    "    .withColumn(\"rank_tpm\",F.lag(\"temp\").over(temp_per_year))\n",
    "    .withColumn(\"rank_tpm2\",F.lag(\"temp\",2).over(temp_per_year)).show()\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "id": "a936fb67-d1a2-42bf-9b54-8e1af7d5bee1",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------+----+---+---+----+----------+------------------+\n",
      "|   stn|year| mo| da|temp|count_temp|      percent_rank|\n",
      "+------+----+---+---+----+----------+------------------+\n",
      "|994979|2017| 12| 11|21.3|        21|0.3333333333333333|\n",
      "|998012|2017| 03| 02|31.4|        24|0.6666666666666666|\n",
      "|719200|2017| 10| 09|60.5|        11|               1.0|\n",
      "|996470|2018| 03| 12|55.6|        12|0.3333333333333333|\n",
      "|076470|2018| 06| 07|65.0|        24|0.6666666666666666|\n",
      "|917350|2018| 04| 21|82.6|         9|               1.0|\n",
      "|041680|2019| 02| 19|16.1|        15|              0.25|\n",
      "|998166|2019| 03| 20|34.8|        12|               0.5|\n",
      "|998252|2019| 04| 18|44.7|        11|              0.75|\n",
      "|949110|2019| 11| 23|54.9|        14|               1.0|\n",
      "+------+----+---+---+----+----------+------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "gsod_light.withColumn(\"percent_rank\",F.cume_dist().over(temp_per_year)).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 24,
   "id": "5ec6dc57-2017-469e-bffc-848dda6c19ac",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------+----+---+---+----+----------+------------------+------------------+\n",
      "|   stn|year| mo| da|temp|count_temp|            avg_NO|             avg_O|\n",
      "+------+----+---+---+----+----------+------------------+------------------+\n",
      "|994979|2017| 12| 11|21.3|        21|37.733333333333334|              21.3|\n",
      "|998012|2017| 03| 02|31.4|        24|37.733333333333334|             26.35|\n",
      "|719200|2017| 10| 09|60.5|        11|37.733333333333334|37.733333333333334|\n",
      "|996470|2018| 03| 12|55.6|        12| 67.73333333333333|              55.6|\n",
      "|076470|2018| 06| 07|65.0|        24| 67.73333333333333|              60.3|\n",
      "|917350|2018| 04| 21|82.6|         9| 67.73333333333333| 67.73333333333333|\n",
      "|041680|2019| 02| 19|16.1|        15|            37.625|              16.1|\n",
      "|998166|2019| 03| 20|34.8|        12|            37.625|             25.45|\n",
      "|998252|2019| 04| 18|44.7|        11|            37.625|31.866666666666664|\n",
      "|949110|2019| 11| 23|54.9|        14|            37.625|            37.625|\n",
      "+------+----+---+---+----+----------+------------------+------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "not_ordered = Window.partitionBy(\"year\")\n",
    "ordered = not_ordered.orderBy(\"temp\")\n",
    "gsod_light.withColumn(\"avg_NO\",F.avg(\"temp\").over(not_ordered)).withColumn(\"avg_O\",F.avg(\"temp\").over(ordered)).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 25,
   "id": "afcfd51d-f77d-491f-99fe-5db84d939f70",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------+----+---+---+----+----------+------------------+------------------+\n",
      "|   stn|year| mo| da|temp|count_temp|            avg_NO|             avg_O|\n",
      "+------+----+---+---+----+----------+------------------+------------------+\n",
      "|994979|2017| 12| 11|21.3|        21|37.733333333333334|              21.3|\n",
      "|998012|2017| 03| 02|31.4|        24|37.733333333333334|             26.35|\n",
      "|719200|2017| 10| 09|60.5|        11|37.733333333333334|37.733333333333334|\n",
      "|996470|2018| 03| 12|55.6|        12| 67.73333333333333|              55.6|\n",
      "|076470|2018| 06| 07|65.0|        24| 67.73333333333333|              60.3|\n",
      "|917350|2018| 04| 21|82.6|         9| 67.73333333333333| 67.73333333333333|\n",
      "|041680|2019| 02| 19|16.1|        15|            37.625|              16.1|\n",
      "|998166|2019| 03| 20|34.8|        12|            37.625|             25.45|\n",
      "|998252|2019| 04| 18|44.7|        11|            37.625|31.866666666666664|\n",
      "|949110|2019| 11| 23|54.9|        14|            37.625|            37.625|\n",
      "+------+----+---+---+----+----------+------------------+------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "not_ordered = Window.partitionBy(\"year\").rowsBetween(Window.unboundedPreceding,Window.unboundedFollowing)\n",
    "ordered = not_ordered.orderBy(\"temp\").rangeBetween(Window.unboundedPreceding,Window.currentRow)\n",
    "gsod_light.withColumn(\"avg_NO\",F.avg(\"temp\").over(not_ordered)).withColumn(\"avg_O\",F.avg(\"temp\").over(ordered)).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 26,
   "id": "55f54fcf-4dfd-495d-b272-40b12029d75a",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------+----+---+---+----+----------+------------------+------------------+\n",
      "|   stn|year| mo| da|temp|count_temp|            avg_NO|             avg_O|\n",
      "+------+----+---+---+----+----------+------------------+------------------+\n",
      "|994979|2017| 12| 11|21.3|        21|37.733333333333334|37.733333333333334|\n",
      "|998012|2017| 03| 02|31.4|        24|37.733333333333334|37.733333333333334|\n",
      "|719200|2017| 10| 09|60.5|        11|37.733333333333334|37.733333333333334|\n",
      "|996470|2018| 03| 12|55.6|        12| 67.73333333333333| 67.73333333333333|\n",
      "|076470|2018| 06| 07|65.0|        24| 67.73333333333333| 67.73333333333333|\n",
      "|917350|2018| 04| 21|82.6|         9| 67.73333333333333| 67.73333333333333|\n",
      "|041680|2019| 02| 19|16.1|        15|            37.625|            37.625|\n",
      "|998166|2019| 03| 20|34.8|        12|            37.625|            37.625|\n",
      "|998252|2019| 04| 18|44.7|        11|            37.625|            37.625|\n",
      "|949110|2019| 11| 23|54.9|        14|            37.625|            37.625|\n",
      "+------+----+---+---+----+----------+------------------+------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "not_ordered = Window.partitionBy(\"year\").rowsBetween(Window.unboundedPreceding,Window.unboundedFollowing)\n",
    "ordered = not_ordered.orderBy(\"temp\").rangeBetween(Window.unboundedPreceding,Window.unboundedFollowing)\n",
    "gsod_light.withColumn(\"avg_NO\",F.avg(\"temp\").over(not_ordered)).withColumn(\"avg_O\",F.avg(\"temp\").over(ordered)).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 27,
   "id": "b51e0162-2fb4-44d5-8a85-9a7631fe299e",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------+----+---+---+----+----------+----------+----------+\n",
      "|   stn|year| mo| da|temp|count_temp|        dt|    dt_num|\n",
      "+------+----+---+---+----+----------+----------+----------+\n",
      "|994979|2019| 12| 11|21.3|        21|2019-12-11|1575993600|\n",
      "|998012|2019| 03| 02|31.4|        24|2019-03-02|1551456000|\n",
      "|719200|2019| 10| 09|60.5|        11|2019-10-09|1570550400|\n",
      "|917350|2019| 04| 21|82.6|         9|2019-04-21|1555776000|\n",
      "|076470|2019| 06| 07|65.0|        24|2019-06-07|1559836800|\n",
      "|996470|2019| 03| 12|55.6|        12|2019-03-12|1552320000|\n",
      "|041680|2019| 02| 19|16.1|        15|2019-02-19|1550505600|\n",
      "|949110|2019| 11| 23|54.9|        14|2019-11-23|1574438400|\n",
      "|998252|2019| 04| 18|44.7|        11|2019-04-18|1555516800|\n",
      "|998166|2019| 03| 20|34.8|        12|2019-03-20|1553011200|\n",
      "+------+----+---+---+----+----------+----------+----------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "gsod_light_p = (\n",
    "    gsod_light\n",
    "    .withColumn(\"year\",F.lit(2019))\n",
    "    .withColumn(\n",
    "        \"dt\",\n",
    "        F.to_date(\n",
    "            F.concat_ws(\"-\",F.col(\"year\"),F.col(\"mo\"),F.col(\"da\"))\n",
    "        )\n",
    "    )\n",
    "    .withColumn(\"dt_num\",F.unix_timestamp(\"dt\"))\n",
    ")\n",
    "gsod_light_p.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 28,
   "id": "283ec130-8a00-4e7d-9d10-c29a09c92902",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------+----+---+---+----+----------+----------+----------+------------------+\n",
      "|   stn|year| mo| da|temp|count_temp|        dt|    dt_num|         avg_count|\n",
      "+------+----+---+---+----+----------+----------+----------+------------------+\n",
      "|041680|2019| 02| 19|16.1|        15|2019-02-19|1550505600|             15.75|\n",
      "|998012|2019| 03| 02|31.4|        24|2019-03-02|1551456000|             15.75|\n",
      "|996470|2019| 03| 12|55.6|        12|2019-03-12|1552320000|             15.75|\n",
      "|998166|2019| 03| 20|34.8|        12|2019-03-20|1553011200|              14.8|\n",
      "|998252|2019| 04| 18|44.7|        11|2019-04-18|1555516800|10.666666666666666|\n",
      "|917350|2019| 04| 21|82.6|         9|2019-04-21|1555776000|              10.0|\n",
      "|076470|2019| 06| 07|65.0|        24|2019-06-07|1559836800|              24.0|\n",
      "|719200|2019| 10| 09|60.5|        11|2019-10-09|1570550400|              11.0|\n",
      "|949110|2019| 11| 23|54.9|        14|2019-11-23|1574438400|              17.5|\n",
      "|994979|2019| 12| 11|21.3|        21|2019-12-11|1575993600|              17.5|\n",
      "+------+----+---+---+----+----------+----------+----------+------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "ONE_MONTH_ISH = 30*60*60*24\n",
    "one_month_ish_before_and_after = (\n",
    "    Window\n",
    "    .partitionBy(\"year\")\n",
    "    .orderBy(\"dt_num\")\n",
    "    .rangeBetween(-ONE_MONTH_ISH,ONE_MONTH_ISH)\n",
    ")\n",
    "gsod_light_p.withColumn(\n",
    "    \"avg_count\",F.avg(\"count_temp\").over(one_month_ish_before_and_after)\n",
    ").show()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "89dcb791-7eac-4d67-93b6-b363f1b7afd6",
   "metadata": {},
   "source": [
    "exercise 1"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 34,
   "id": "290c8bd1-8226-4aac-ab56-c3291089b8eb",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---+---+-----+\n",
      "|ord|row|range|\n",
      "+---+---+-----+\n",
      "| 10|  3|  101|\n",
      "| 10|  4|  101|\n",
      "| 10|  5|  101|\n",
      "| 10|  5|  101|\n",
      "| 10|  5|  101|\n",
      "| 10|  5|  101|\n",
      "| 10|  5|  101|\n",
      "| 10|  5|  101|\n",
      "| 10|  5|  101|\n",
      "| 10|  5|  101|\n",
      "+---+---+-----+\n",
      "only showing top 10 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "ord_df = spark.createDataFrame([[10] for _ in range(101)],[\"ord\"])\n",
    "ord_df.select(\n",
    "    \"ord\",\n",
    "    F.count(\"ord\").over(Window.partitionBy().orderBy(\"ord\").rowsBetween(-2,2)).alias(\"row\"),\n",
    "    F.count(\"ord\").over(Window.partitionBy().orderBy(\"ord\").rangeBetween(-2,2)).alias(\"range\")\n",
    ").show(10)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "25b76409-a58d-4739-8651-2e250a51b76a",
   "metadata": {},
   "source": [
    "exercise 2"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 32,
   "id": "a57fe505-fd09-4f37-a37c-c6c506b90c50",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------+----+---+---+------+--------+\n",
      "|   stn|year| mo| da|  temp|avg_temp|\n",
      "+------+----+---+---+------+--------+\n",
      "|896250|2017| 06| 20|-114.7|  -114.7|\n",
      "|896060|2018| 08| 27|-113.5|  -113.5|\n",
      "|895770|2019| 06| 15|-114.7|  -114.7|\n",
      "+------+----+---+---+------+--------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "(gsod\n",
    " .withColumn(\"min_temp\",F.min(\"temp\").over(each_year))\n",
    " .where(\"min_temp=temp\")\n",
    " .select(\"stn\",\"year\",\"mo\",\"da\",\"temp\")\n",
    " .orderBy(\"year\",\"mo\",\"da\")\n",
    " .withColumn(\"avg_temp\",F.avg(\"temp\").over(each_year))\n",
    " .orderBy(\"year\",\"mo\",\"da\")\n",
    " .show()\n",
    ")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "6e788320-fb7b-47df-8c6f-ca35bef6aad1",
   "metadata": {},
   "source": [
    "exercise 3"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 33,
   "id": "70243fa5-3a16-4d3d-b19b-071dfce47a25",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------+----+---+---+----+----------+--------+\n",
      "|   stn|year| mo| da|temp|count_temp|rank_tpm|\n",
      "+------+----+---+---+----+----------+--------+\n",
      "|041680|2019| 02| 19|16.1|        15|       1|\n",
      "|996470|2018| 03| 12|55.6|        12|       1|\n",
      "|998166|2019| 03| 20|34.8|        12|       2|\n",
      "|998012|2017| 03| 02|31.4|        24|       3|\n",
      "|917350|2018| 04| 21|82.6|         9|       1|\n",
      "|998252|2019| 04| 18|44.7|        11|       2|\n",
      "|076470|2018| 06| 07|65.0|        24|       1|\n",
      "|719200|2017| 10| 09|60.5|        11|       1|\n",
      "|949110|2019| 11| 23|54.9|        14|       1|\n",
      "|994979|2017| 12| 11|21.3|        21|       1|\n",
      "+------+----+---+---+----+----------+--------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "temp_per_month_asc = Window.partitionBy(\"mo\").orderBy(\"count_temp\")\n",
    "gsod_light.withColumn(\"rank_tpm\",F.row_number().over(temp_per_month_asc)).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 36,
   "id": "60bfaf23-173b-4685-83ff-d387744821f2",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------+----+---+---+----+\n",
      "|   stn|year| mo| da|temp|\n",
      "+------+----+---+---+----+\n",
      "|994979|2017| 12| 11|21.3|\n",
      "|998012|2017| 03| 02|31.4|\n",
      "|719200|2017| 10| 09|60.5|\n",
      "|998258|2017| 05| 01|44.9|\n",
      "|997737|2017| 03| 10| 6.8|\n",
      "|025010|2017| 11| 24|44.8|\n",
      "|992190|2017| 10| 08|54.6|\n",
      "|719200|2017| 09| 16|58.5|\n",
      "|997356|2017| 07| 10|81.9|\n",
      "|997737|2017| 10| 12|55.4|\n",
      "+------+----+---+---+----+\n",
      "only showing top 10 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "(gsod\n",
    " .withColumn(\"min_temp\",F.min(\"temp\").over(each_year))\n",
    " .select(\"stn\",\"year\",\"mo\",\"da\",\"temp\")\n",
    " .show(10)\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 39,
   "id": "0dfee9f9",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Collecting pyarrow\n",
      "  Downloading pyarrow-18.1.0-cp39-cp39-win_amd64.whl (25.3 MB)\n",
      "Installing collected packages: pyarrow\n",
      "Successfully installed pyarrow-18.1.0\n"
     ]
    }
   ],
   "source": [
    "# !pip install pyarrow"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 41,
   "id": "b95a6711",
   "metadata": {},
   "outputs": [],
   "source": [
    "import pandas as pd\n",
    "from functools import map"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 68,
   "id": "7785d260",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------+----------+----+----------+\n",
      "|   stn|        ut|temp|7_max_temp|\n",
      "+------+----------+----+----------+\n",
      "|010060|1483286400|20.2|      25.7|\n",
      "|010060|1483372800|17.9|      25.7|\n",
      "|010060|1483459200|19.5|      25.7|\n",
      "|010060|1483545600|25.7|      25.7|\n",
      "|010060|1483632000|21.1|      25.7|\n",
      "|010060|1483718400|19.1|      25.7|\n",
      "|010060|1483804800|13.5|      25.7|\n",
      "|010060|1483891200|10.3|      27.8|\n",
      "|010060|1483977600| 6.0|      27.8|\n",
      "|010060|1484064000| 7.0|      27.8|\n",
      "+------+----------+----+----------+\n",
      "only showing top 10 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "SEVEN_DAY_TIME = 7*24*60*60\n",
    "(gsod\n",
    " .withColumn(\"ut\",F.unix_timestamp(F.to_date(F.concat_ws(\"-\",F.col(\"year\"),F.col(\"mo\"),F.col(\"da\")))))\n",
    " .orderBy(\"ut\")\n",
    " .withColumn(\"7_max_temp\",\n",
    "             F.max(\"temp\")\n",
    "             .over(Window\n",
    "                   .partitionBy(\"stn\")\n",
    "                   .orderBy(\"ut\")\n",
    "                   .rangeBetween(-SEVEN_DAY_TIME,SEVEN_DAY_TIME)\n",
    "                  )\n",
    "            )\n",
    " .select(\"stn\",\"ut\",\"temp\",\"7_max_temp\")\n",
    " .show(10)\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 70,
   "id": "6c0e39ee",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------+----------+----+----------+-------------+\n",
      "|   stn|        ut|temp|7_max_temp|is_7_max_temp|\n",
      "+------+----------+----+----------+-------------+\n",
      "|010060|1483286400|20.2|      25.7|        false|\n",
      "|010060|1483372800|17.9|      25.7|        false|\n",
      "|010060|1483459200|19.5|      25.7|        false|\n",
      "|010060|1483545600|25.7|      25.7|         true|\n",
      "|010060|1483632000|21.1|      25.7|        false|\n",
      "|010060|1483718400|19.1|      25.7|        false|\n",
      "|010060|1483804800|13.5|      25.7|        false|\n",
      "|010060|1483891200|10.3|      27.8|        false|\n",
      "|010060|1483977600| 6.0|      27.8|        false|\n",
      "|010060|1484064000| 7.0|      27.8|        false|\n",
      "+------+----------+----+----------+-------------+\n",
      "only showing top 10 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "@F.pandas_udf(T.BooleanType())\n",
    "def is_7_max_temp(temps:pd.Series,max_temps:pd.Series)->pd.Series:\n",
    "    return temps == max_temps\n",
    "\n",
    "SEVEN_DAY_TIME = 7*24*60*60\n",
    "(gsod\n",
    " .withColumn(\"ut\",F.unix_timestamp(F.to_date(F.concat_ws(\"-\",F.col(\"year\"),F.col(\"mo\"),F.col(\"da\")))))\n",
    " .orderBy(\"ut\")\n",
    " .withColumn(\"7_max_temp\",\n",
    "             F.max(\"temp\")\n",
    "             .over(Window\n",
    "                   .partitionBy(\"stn\")\n",
    "                   .orderBy(\"ut\")\n",
    "                   .rangeBetween(-SEVEN_DAY_TIME,SEVEN_DAY_TIME)\n",
    "                  )\n",
    "            )\n",
    " .withColumn(\"is_7_max_temp\",is_7_max_temp(F.col(\"temp\"),F.col(\"7_max_temp\")))\n",
    " .select(\"stn\",\"ut\",\"temp\",\"7_max_temp\",\"is_7_max_temp\")\n",
    " .show(10)\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 59,
   "id": "7e4dcbfd",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "array([ True, False, False])"
      ]
     },
     "execution_count": 59,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "a = pd.Series([1,2,3])\n",
    "(a == 1).values"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 73,
   "id": "777788bc",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- stn: string (nullable = true)\n",
      " |-- wban: string (nullable = true)\n",
      " |-- year: string (nullable = true)\n",
      " |-- mo: string (nullable = true)\n",
      " |-- da: string (nullable = true)\n",
      " |-- temp: double (nullable = true)\n",
      " |-- count_temp: long (nullable = true)\n",
      " |-- dewp: double (nullable = true)\n",
      " |-- count_dewp: long (nullable = true)\n",
      " |-- slp: double (nullable = true)\n",
      " |-- count_slp: long (nullable = true)\n",
      " |-- stp: double (nullable = true)\n",
      " |-- count_stp: long (nullable = true)\n",
      " |-- visib: double (nullable = true)\n",
      " |-- count_visib: long (nullable = true)\n",
      " |-- wdsp: string (nullable = true)\n",
      " |-- count_wdsp: string (nullable = true)\n",
      " |-- mxpsd: string (nullable = true)\n",
      " |-- gust: double (nullable = true)\n",
      " |-- max: double (nullable = true)\n",
      " |-- flag_max: string (nullable = true)\n",
      " |-- min: double (nullable = true)\n",
      " |-- flag_min: string (nullable = true)\n",
      " |-- prcp: double (nullable = true)\n",
      " |-- flag_prcp: string (nullable = true)\n",
      " |-- sndp: double (nullable = true)\n",
      " |-- fog: string (nullable = true)\n",
      " |-- rain_drizzle: string (nullable = true)\n",
      " |-- snow_ice_pellets: string (nullable = true)\n",
      " |-- hail: string (nullable = true)\n",
      " |-- thunder: string (nullable = true)\n",
      " |-- tornado_funnel_cloud: string (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "gsod.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 74,
   "id": "1a72401c",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------+----+---+---+----+----------+\n",
      "|   stn|year| mo| da|temp|count_temp|\n",
      "+------+----+---+---+----+----------+\n",
      "|994979|2017| 12| 11|21.3|        21|\n",
      "|998012|2017| 03| 02|31.4|        24|\n",
      "|719200|2017| 10| 09|60.5|        11|\n",
      "|917350|2018| 04| 21|82.6|         9|\n",
      "|076470|2018| 06| 07|65.0|        24|\n",
      "|996470|2018| 03| 12|55.6|        12|\n",
      "|041680|2019| 02| 19|16.1|        15|\n",
      "|949110|2019| 11| 23|54.9|        14|\n",
      "|998252|2019| 04| 18|44.7|        11|\n",
      "|998166|2019| 03| 20|34.8|        12|\n",
      "+------+----+---+---+----+----------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "gsod_light.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 80,
   "id": "08867b6c",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------+----+---+---+----+----------+------------------+\n",
      "|   stn|year| mo| da|temp|count_temp|          avg_temp|\n",
      "+------+----+---+---+----+----------+------------------+\n",
      "|041680|2019|  2| 19|16.1|        15|34.474999999999994|\n",
      "|998012|2019|  3| 02|31.4|        24|44.199999999999996|\n",
      "|996470|2019|  3| 12|55.6|        12|44.199999999999996|\n",
      "|998166|2019|  3| 20|34.8|        12|44.199999999999996|\n",
      "|917350|2019|  4| 21|82.6|         9| 49.81999999999999|\n",
      "|998252|2019|  4| 18|44.7|        11| 49.81999999999999|\n",
      "|076470|2019|  6| 07|65.0|        24|              65.0|\n",
      "|719200|2019| 10| 09|60.5|        11|              57.7|\n",
      "|949110|2019| 11| 23|54.9|        14| 45.56666666666667|\n",
      "|994979|2019| 12| 11|21.3|        21|              38.1|\n",
      "+------+----+---+---+----+----------+------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "@F.pandas_udf(T.IntegerType())\n",
    "def change_mo(mo:pd.Series)->pd.Series:\n",
    "    return mo.apply(lambda x:int(x.strip(\"0\")) if x.startswith(\"0\") else int(x))\n",
    "\n",
    "gsod_light_p = (\n",
    "    gsod_light\n",
    "    .withColumn(\"year\",F.lit(2019))\n",
    "    .withColumn(\"mo\",F.col(\"mo\").cast(\"int\"))#change_mo(\"mo\")\n",
    "    .withColumn('avg_temp',F.avg(\"temp\").over(Window.partitionBy(\"year\")\n",
    "                                          .orderBy(\"mo\")\n",
    "                                          .rangeBetween(-1,1)\n",
    "                                         )\n",
    "               )\n",
    ")\n",
    "gsod_light_p.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 81,
   "id": "e2644cc2",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "34.474999999999994"
      ]
     },
     "execution_count": 81,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "(16.1+31.4+55.6+34.8)/4"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "0639abe6",
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3 (ipykernel)",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.9.7"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}
